/**
 * 
 */
package config;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import mapper.GraphMapper;

import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputLogFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import partitioning.GraphPartition;
import partitioning.PartitionController;
import partitioning.TCPListener;
import reducer.GraphReducer;
import util.ArrayUtil;
import util.HDFSFileUtil;

/**
 * @author puneet
 * 
 */
public class ShortestPathJobRunnerWithPartitioning {
	private static Pattern nodeMsgPattern = Pattern.compile("^node:\\s(.+?)\\s(.*)$");
	private static Random r = new Random();
	private static Set<String> querySet = new HashSet<String>();
	public static void main(String args[]) throws Exception {
		Option sourceOption = new Option("s", "source", true, "Source Node");
		Option destOption = new Option("d", "dest", true, "Source Node");
		Option serversOption = new Option("os", "servers", true, "Other servers in the form hostname:port");
		Option listenPortOption = new Option("l", "listen", true, "Which port to listen on");
		Option startRunningOption = new Option("sr", "start-running", true, "Start running? yes/no");
		Option inputPartitionFileOption = new Option("ipf", "input-file", true, "Input Partition File on HDFS");
		Option mapInputDirOption = new Option("mi", "map-input-dir", true, "Map Input Directory");
		Option outputDirOption = new Option("od", "output-dir", true, "Output Directory");
		Option nodeIdOption = new Option("ni","node-id",false,"Node Id (Default is hostname)");
		
		Options options = new Options();
		options.addOption(sourceOption);
		options.addOption(destOption);
		options.addOption(serversOption);
		options.addOption(listenPortOption);
		options.addOption(inputPartitionFileOption);
		options.addOption(mapInputDirOption);
		options.addOption(outputDirOption);
		options.addOption(startRunningOption);
		options.addOption(nodeIdOption);
		Configuration conf = new Configuration();

		GenericOptionsParser optionsParser = new GenericOptionsParser(conf, options, args);

		String source = optionsParser.getCommandLine().getOptionValue("s");
		String dest = optionsParser.getCommandLine().getOptionValue("d");
		String otherServersString = optionsParser.getCommandLine().getOptionValue("os");
		int listenPort = Integer.parseInt(optionsParser.getCommandLine().getOptionValue("l"));
		// Register partitions
		registerPartitions(listenPort, otherServersString);
		System.out.println("Registered Partitions");
		// Build nodes neighbor map
		FileSystem fs = FileSystem.get(conf);
		Map<String, String[]> nodesNeighborsMap = buildNodeNeighborsMap(optionsParser.getCommandLine().getOptionValue("ipf"), fs);
		System.out.println("Built node Neighbor Map");
		String startRunningString = optionsParser.getCommandLine().getOptionValue("sr");
		String restOfNodes=null;
		boolean startRunning = false;
		if ("yes".equalsIgnoreCase(startRunningString)) {
			startRunning = true;
			restOfNodes = source + ":"+0;
		}
		String nodeId = optionsParser.getCommandLine().getOptionValue("ni",InetAddress.getLocalHost().getHostName());
		String mapInputDir = optionsParser.getCommandLine().getOptionValue("mi");
		String outputDir = optionsParser.getCommandLine().getOptionValue("od");
		String mapFilePrefix = mapInputDir+"/map-"+nodeId+"-";
		TCPListener listener = new TCPListener(listenPort);
		String nodeQuery=null;
		
		mainloop:while (true) {
			if (!startRunning) {
				String msg = listener.blockWaitForMessage().trim();
				if(msg == null) {
					System.out.println("Null message received");
					continue mainloop;
				}
				Matcher m = nodeMsgPattern.matcher(msg);
				if (m.matches()) {
					nodeQuery = m.group(1);
					if(querySet.contains(nodeQuery)) {
						System.err.println("Duplicate query. Ending. Node - "+nodeQuery);
						//return;
					}
					querySet.add(nodeQuery);
					restOfNodes = m.group(2);
					if (nodesNeighborsMap.containsKey(nodeQuery)) {
						System.out.println("Found this node "+nodeQuery+" here.");
						startRunning = true;
					}else{
						System.out.println("Node query for:" + nodeQuery + " Not found at node "+nodeId);
					}
				}
				else if(msg.startsWith("end")) {
					System.out.println("End message received");
					break;
				}else {
					continue;
				}
				
			}else {
				System.out.println("Rest of Nodes String:"+restOfNodes);
				//Transfer of control or this was the starting
				Map<String, Integer> nodeDistanceMap = new TreeMap<String, Integer>();
				for(String nodeDistancePair:restOfNodes.split("\\s")) {
					String[] nodeDistance = nodeDistancePair.split(":");
					nodeDistanceMap.put(nodeDistance[0], Integer.parseInt(nodeDistance[1]));
				}
				System.out.println("Node Distance Map is:" + nodeDistanceMap);
				Path mapFilePath = null;
				while(true) {
					
					mapFilePath = new Path(mapFilePrefix+r.nextInt());
					if(!fs.exists(mapFilePath)) {
						break;
					}
				}
				BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs.create(mapFilePath)));
				
				for(Iterator<String> nodeSetIterator = nodeDistanceMap.keySet().iterator();nodeSetIterator.hasNext();) {
					String node = nodeSetIterator.next();
					if(node.equalsIgnoreCase(dest)) {
						System.out.println("Found dest at distance:"+nodeDistanceMap.get(node));
						PartitionController.broadcastMessage("end");
						bw.close();
						break mainloop;
					}
					if(!nodesNeighborsMap.containsKey(node)) {
						System.out.println("Node miss on partition "+ nodeId+ " for node"+node + " Now transferring control");
						startRunning = false;
						String msgToBroadcast = "node: "+node+" "+nodeDistanceMapSerializer(nodeDistanceMap)+"\n";
						System.out.println("Msg to be broadcasted:" + msgToBroadcast);
						PartitionController.broadcastMessage(msgToBroadcast);
						bw.close();
						continue mainloop;
					}else {
						bw.write(node+" "+nodeDistanceMap.get(node)+" "+ArrayUtil.join(nodesNeighborsMap.get(node), " ")+"\n");						
						nodeSetIterator.remove();
					}
				}
				bw.flush();
				bw.close();
				//Now run the hadoop job.
				System.out.println("Finished here. Now running hadoop job here");
				Job job = new Job(conf, "graph search map reduce");
				job.setMapperClass(GraphMapper.class);
				
				job.setReducerClass(GraphReducer.class);
				job.setCombinerClass(GraphReducer.class);
				job.setJarByClass(ShortestPathJobRunnerWithPartitioning.class);
				job.setOutputKeyClass(Text.class);
				job.setOutputValueClass(IntWritable.class);

				MyInputFormat.addInputPath(job, new Path(mapInputDir));
				Path outputDirPath = new Path(outputDir);
				if (fs.exists(outputDirPath)) {
					fs.delete(outputDirPath, true);
				}
				FileOutputFormat.setOutputPath(job, outputDirPath);
				job.waitForCompletion(true);			
				Path[] outputFileList = FileUtil.stat2Paths(fs.listStatus(outputDirPath,new OutputLogFilter()));
				StringBuffer restOfNodesBufferString = new StringBuffer();
				for(Path opfile:outputFileList) {
					BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(opfile)));
					String s = null;
					while((s=br.readLine()) != null) {
						String[] vals = s.split("\\s");
						if(dest.equals(vals[0])) {
							br.close();
							System.out.println("Shortest Distance found:"+vals[1]);
							PartitionController.broadcastMessage("end "+vals[1]);
						}
						restOfNodesBufferString.append(vals[0]+":" + vals[1]+" ");
					}
					restOfNodes = restOfNodesBufferString.toString().trim();
					br.close();
				}
				System.out.println("New rest of nodes = "+ restOfNodes);
				//Now delete all map files
				System.out.println("Deleting map files");
				Path[] mapFilesList=FileUtil.stat2Paths(fs.listStatus(new Path(mapInputDir)));
				for(Path mfile:mapFilesList) {
					fs.delete(mfile, false);
				}
			}
		}

		System.out.println("Source = " + source + " Dest=" + dest);
	}

	private static Map<String, String[]> buildNodeNeighborsMap(String inputfilePath, FileSystem fs) {
		List<String> lines = HDFSFileUtil.readFileAsLineList(inputfilePath, fs);
		Map<String, String[]> nodesNeighborsMap = new HashMap<String, String[]>();
		for (String line : lines) {
			String[] vals = line.split("\\s");
			nodesNeighborsMap.put(vals[0], ArrayUtil.splice(vals, 1, vals.length));
		}
		return nodesNeighborsMap;
	}

	private static void registerPartitions(int localPort, String otherServersString) {
		for (String serverPortPair : otherServersString.split(",")) {
			String[] serverPort = serverPortPair.split(":");
			GraphPartition gp = new GraphPartition();
			gp.setHost(serverPort[0]);
			gp.setLocal(false);
			gp.setPort(Integer.parseInt(serverPort[1]));
			PartitionController.registerPartition(gp);
		}
		GraphPartition localPartition = new GraphPartition();
		localPartition.setHost("localhost");
		localPartition.setLocal(true);
		localPartition.setPort(localPort);
		PartitionController.registerPartition(localPartition);
	}
	
	private static  String nodeDistanceMapSerializer(Map<String,Integer> nodeDistanceMap) {
		StringBuffer buffer = new StringBuffer();
		for(Entry<String, Integer> nodeDistanceCombination:nodeDistanceMap.entrySet()) {
			buffer.append(nodeDistanceCombination.getKey()+":"+nodeDistanceCombination.getValue()+" ");
		}
		buffer.setLength(buffer.length()-1);
		return buffer.toString();
	}
}
